package com.boarsoft.message.rocketmq;

import java.io.Closeable;
import java.util.Map;

import javax.annotation.PostConstruct;

import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 基于RocketMQ客户端完成消息的外发<br>
 * 应用可以直接使用RocketMQ的API完成外发，也可以通过本类来完成<br>
 * 将RocketMQClient的实例，按MessageService接口注入Messenger即可。
 * 
 * @author Mac_J
 *
 */
public class RocketMQClient implements Closeable {
	private static final Logger log = LoggerFactory.getLogger(RocketMQClient.class);

	/** 队列列表 */
	protected Map<String, RocketMQDestination> destMap;

	@PostConstruct
	public void init() {
		for (String key : destMap.keySet()) {
			RocketMQDestination dest = destMap.get(key);
			dest.setCode(key);
			// 初始化此队列的消费者
			RocketMQConsumer consumer = dest.getConsumer();
			if (consumer != null) {
				consumer.init(dest.getCode());
			}
			// 初始化此队列的生产者
			RocketMQProducer producer = dest.getProducer();
			if (producer != null) {
				producer.init();
			}
		}
	}

	/**
	 * 外发消息
	 * 
	 * @param msg
	 * @param target
	 * @throws Exception
	 */
	public void put(Message m, String target) throws Exception {
		RocketMQDestination dest = destMap.get(target);
		RocketMQProducer producer = dest.getProducer();
		SendResult sr = producer.send(m);
		log.info("Send message = {}", sr);
		if (sr.getSendStatus().equals(SendStatus.SEND_OK)) {
			return;
		}
		throw new IllegalStateException(String.format(//
				"Send message failed: %s", sr.getSendStatus().toString()));
	}

	@Override
	public void close() {
		for (RocketMQDestination dest : destMap.values()) {
			try {
				dest.close();
			} catch (Exception e) {
				log.error("Error on close RocketMQ destination {}", dest, e);
			}
		}
	}

	public void setDestMap(Map<String, RocketMQDestination> destMap) {
		this.destMap = destMap;
	}
}